package com.myCompany.edu.realtime.app.dwd.db;

import com.myCompany.edu.realtime.app.BaseSQLApp;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.util.FlinkSinkUtil;
import com.myCompany.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd05_interaction_review extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd05_interaction_review().init(2005,2,"Dwd05_interaction_review");
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1.读取ods_db
        readOdsDb(tEnv,"Dwd05_interaction_review");

        // 2.过滤出评价数据。insert && edu && delited=0
        Table reviewInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id ," +
                " `data`['user_id'] user_id ," +
                " `data`['course_id'] course_id ," +
                " cast(`data`['review_stars'] as bigint) review_stars ," +
                " `data`['create_time'] create_time ," +
                "  ts " +
                " from ods_db " +
                " where `database` = 'edu' " +
                " and   `table` = 'review_info' " +
                " and `type` = 'insert' " +
                " and `data`['deleted'] = '0' ");
    tEnv.createTemporaryView("review_info",reviewInfo);

    tEnv.executeSql("create table dwd_interaction_review (" +
            " id           string ," +
            " user_id      string ," +
            " course_id    string ," +
            " review_stars bigint ," +
            " create_time  string ," +
            " ts bigint )" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_INTERACTION_REVIEW)
    );
    reviewInfo.executeInsert("dwd_interaction_review");



    }
}
